package com.example.kafka.listener;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import jdk.nashorn.internal.ir.ObjectNode;
import lombok.Getter;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.TopicPartitionOffset;

import java.util.Properties;

/**
 *
 */
@Slf4j
@Getter
public class IOTListener implements AcknowledgingMessageListener<String, String> {

    private static final String TOPIC = "DaHua-local-test";

    private ObjectMapper objectMapper;

    private String groupId;

    public IOTListener(ObjectMapper objectMapper,String groupId){
        this.objectMapper = objectMapper;
        this.groupId = groupId;
    }

    @Override
    public void onMessage(@NonNull ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        ObjectNode node = null;
        log.info("消息来源分区: {}", record.partition());
        try {
            node = objectMapper.readValue(record.value(), ObjectNode.class);
        } catch (JsonProcessingException e) {
            log.error("字符串转json对象失败 : {}", record.value());
            return;
        }finally {
            acknowledgment.acknowledge();
        }
        log.info("字符串转json完成 : {}", node.toString());
    }

    /**
     * 监听器默认配置
     * 需要配置监听的主题 监听的分区
     * @param groupId
     * @param topic
     * @param partition
     * @return
     */
    public ContainerProperties getContainerProperties(String groupId, String topic, int partition){
        if (StringUtils.isBlank(topic)){
            topic = TOPIC;
            partition = 0;
        }
        TopicPartitionOffset tpo = new TopicPartitionOffset(topic, partition);
        ContainerProperties properties = new ContainerProperties(tpo);
        /**
         * 设置ACK模式(手动提交模式，这里有七种)
         * RECORD： 每处理完一条记录后提交。
         * BATCH(默认)： 每次poll一批数据后提交一次，频率取决于每次poll的调用频率。
         * TIME： 每次间隔ackTime的时间提交。
         * COUNT： 处理完poll的一批数据后并且距离上次提交处理的记录数超过了设置的ackCount就提交。
         * COUNT_TIME： TIME和COUNT中任意一条满足即提交。
         * MANUAL： 手动调用Acknowledgment.acknowledge()后，并且处理完poll的这批数据后提交。
         * MANUAL_IMMEDIATE： 手动调用Acknowledgment.acknowledge()后立即提交。
         * 如果设置 AckMode 模式为 MANUAL 或者 MANUAL_IMMEDIATE，则需要对监听消息的方法中，引入 Acknowledgment 对象参数，并调用 acknowledge() 方法进行手动提交
         *
         */
        properties.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        properties.setGroupId(groupId);
        //设置其他属性
        Properties consumerProperties = properties.getKafkaConsumerProperties();
        consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
        consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
        //当前对象设置到 properties 中去
        properties.setMessageListener(this);
        return properties;
    }
}
